package com.zhao.sparkstreaming.sgg.code.handler

import java.sql.{Connection, Date}
import java.text.SimpleDateFormat

import com.zhao.sparkstreaming.sgg.code.Ads_log
import com.zhao.sparkstreaming.sgg.utils.JdbcUtil
import org.apache.spark.streaming.dstream.DStream

/**
 * Description: 实现实时的动态黑名单机制:将每天对某个广告点击超过100次的用户拉黑<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/4 15:58
 * 思路分析:
 * 1)读取Kafka数据之后,并对Mysql中存储的黑名单数据做校验
 * 2)校验通过则对用户点击广告次数累加1并存入Mysql
 * 3)在存入Mysql之后对数据做校验,如果单日超过100次则将该用户加入黑名单
 *
 * @author 柒柒
 * @version : 1.0
 */

object BlackListHandler {

  //时间格式化对象
  private val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")

  def addBlackList(filterAdsLogDSteam: DStream[Ads_log])= {

    //统计当前批次中单日每个用户名点击每个广告的总次数
    //1.将数据转换结构 ads_log=>((date,user,adid),1)
    val dateUserAdToOne: DStream[((String, String, String), Long)] = filterAdsLogDSteam.map(adsLog => {
      //a.将时间戳转换为日期字符串
      val date: String = sdf.format(new Date(adsLog.timestamp))
      //b.返回值
      ((date, adsLog.userid, adsLog.adid), 1L)
    })

    //2.统计单日每个用户点击每个广告的总次数
    // ((date,user,adid),1)=>((date,user,adid),count)
    val dateUserAdToCount: DStream[((String, String, String), Long)] = dateUserAdToOne.reduceByKey(_ + _)

    dateUserAdToCount.foreachRDD(rdd => {
      rdd.foreachPartition(iter => {
        val connection: Connection = JdbcUtil.getConnection

        iter.foreach { case ((dt, user, ad), count) =>
          JdbcUtil.executeUpdate(connection,
            """
              |insert into user_ad_count(dt,userid,adid,count)
              |values(?,?,?,?)
              |on duplicate key
              |update count=count+?
              |""".stripMargin, Array(dt, user, ad, count, count))

          val ct: Long = JdbcUtil.getDataFromMysql(connection, "select count from user_ad_count where dt=? and userid=? and adid=?",
            Array(dt, user, ad))
          if (ct >= 100) {
            JdbcUtil.executeUpdate(connection, "insert into black_list(userid) value (?) on duplicate key update userid=?",
              Array(user, user))
          }
        }
        connection.close()
      })
    })
  }

  def filterByBlackList(adsLogDStream: DStream[Ads_log]) = {
    adsLogDStream.transform(rdd => {
      rdd.filter(adsLog => {
        val connection: Connection = JdbcUtil.getConnection
        val bool: Boolean = JdbcUtil.isExist(connection, "select * from black_list where userid=?", Array(adsLog.userid))
        connection.close()
        !bool
      })
    })
  }
}
